package com.gws.ievent.scheduler.timewheel;

import java.util.HashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 时间轮
 *
 * @author song
 * @date 2022/10/26 11:26
 */
public class TimeWheel {

    private static final int STATE_INIT = 0;
    private static final int STATE_RUN = 1;
    private static final int STATE_STOP = 2;

    /**
     * 每个槽持续的时间/纳秒
     */
    private final long duration;

    /**
     * 槽位数组
     */
    private final Seat[] seats;

    /**
     * 槽位数组长度-1, tick&mask定位数组下标
     */
    private final int mask;

    /**
     * 时间轮转动的次数
     */
    private final AtomicLong point = new AtomicLong(0);

    /**
     * 状态
     */
    private volatile int state = STATE_INIT;
    private static final AtomicIntegerFieldUpdater<TimeWheel> STATE_UPDATER = AtomicIntegerFieldUpdater.newUpdater(TimeWheel.class, "state");

    /**
     * 开始时间
     */
    private volatile long startTime;

    /**
     * 工作线程
     */
    private final Thread workerThread;

    /**
     * 运行的task数量
     */
    private final AtomicLong appends = new AtomicLong(0);


    /**
     * @param num      时间轮的槽位数
     * @param duration 每个槽位的持续时间
     */
    public TimeWheel(int num, long duration, TimeUnit unit) {
        if (num <= 0) {
            throw new IllegalArgumentException("num: " + num + " (expected: > 0)");
        }

        if (duration <= 0) {
            throw new IllegalArgumentException("duration: " + duration + " (expected: > 0)");
        }

        if (unit == null) {
            throw new NullPointerException("unit");
        }

        long durationNanos = unit.toNanos(duration);
        long oneMills = TimeUnit.MILLISECONDS.toNanos(1);
        if (durationNanos < oneMills) {
            throw new IllegalArgumentException(String.format("Configured tickDuration %d smaller then %d, using 1ms.", duration, oneMills));
        }

        this.seats = this.createSeats(num);
        if (durationNanos >= Long.MAX_VALUE / this.seats.length) {
            throw new IllegalArgumentException(String.format(
                    "tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
                    duration, Long.MAX_VALUE / this.seats.length));
        }

        this.duration = durationNanos;
        this.mask = this.seats.length - 1;
        this.workerThread = Executors.defaultThreadFactory().newThread(new Worker());
    }


    /**
     * 添加任务
     *
     * @param task
     * @param delay
     * @param unit
     */
    public void addTask(Task task, long delay, TimeUnit unit) {
        if (task == null) {
            throw new NullPointerException("task");
        }

        if (delay <= 0) {
            throw new IllegalArgumentException("delay: " + delay + " (expected: > 0)");
        }

        if (unit == null) {
            throw new NullPointerException("unit");
        }

        // 启动时间轮工作线程
        this.start();

        long deadline = System.nanoTime() + unit.toNanos(delay) - this.startTime;
        long numSeat = deadline / this.duration;

        int round = (int) (numSeat / this.seats.length);
        int idx = (int) (numSeat & this.mask);

        this.seats[idx].getEnqueue().add(new Bucket(round, task));
    }


    /**
     * 开始
     */
    private void start() {
        switch (STATE_UPDATER.get(this)) {
            case STATE_INIT:
                boolean b = STATE_UPDATER.compareAndSet(this, STATE_INIT, STATE_RUN);
                if (b) {
                    this.startTime = System.nanoTime();
                    System.out.println(this.startTime);
                    this.workerThread.start();
                }
                break;
            case STATE_RUN:
                break;
            case STATE_STOP:
                throw new IllegalStateException("cannot be started once stopped");
            default:
                throw new Error("Invalid TimeWheel state");
        }
    }


    /**
     * 创建槽位数组
     *
     * @param num
     * @return
     */
    private Seat[] createSeats(int num) {
        int n = num - 1;
        n |= n >>> 1;
        n |= n >>> 2;
        n |= n >>> 4;
        n |= n >>> 8;
        n |= n >>> 16;

        int max = 1 << 30;
        n = (n < 0) ? 1 : ((n >= max) ? max : (n + 1));

        Seat[] seats = new Seat[n];
        for (int i = 0; i < n; ++i) {
            seats[i] = new Seat();
        }
        return seats;
    }


    public long getRunTasks() {
        return this.appends.get();
    }


    private final class Worker implements Runnable {

        @Override
        public void run() {
            System.out.println("state: " + STATE_UPDATER.get(TimeWheel.this));

            while (STATE_UPDATER.get(TimeWheel.this) == STATE_RUN) {
                // 当前索引
                int idx = (int) point.get() & mask;

                // 把槽位中的队列数据转移到该槽位的链表中
                this.transformNextSeatBucket(idx, 10000);

                // 循环执行
                this.taskLoopRun(idx);

                // 指针加1
                point.incrementAndGet();

                // 等待时间的到来
                this.waitForNext();
            }
        }


        /**
         * 等待下一个槽位的到来
         */
        private void waitForNext() {
            long p = point.get();
            long current = duration * (p + 1);

            // 在等待下一个时间点的到来的时候, 把后面槽的队列的数据转移到槽的链表中去
            long max = p + seats.length;
            while (true) {
                long nano = System.nanoTime() - startTime;
                if (nano >= current) {
                    return;
                }

                // 尝试把下一个槽的任务队列放入槽的链表尾部
                boolean b = this.transformNextSeatBucket((int)(p & mask), 100);
                if (b) {
                    ++p;
                }

                // 如果经过一轮循环之后还是没有数据,直接sleep避免无效空轮询
                if (p > max) {
                    try {
                        long sleepNanos = current - nano;
                        long sleepMills = TimeUnit.NANOSECONDS.toMillis(sleepNanos);
                        Thread.sleep(sleepMills);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }


        /**
         * 把下一个槽位中的队列数据转移到该槽位的链表中
         *
         * @param idx
         * @param i
         */
        private boolean transformNextSeatBucket(int idx, int i) {
            Seat seat = seats[idx];

            // 把槽位中的队列数据转移到该槽位的链表中
            return seat.transformEnqueueToLink(i);
        }


        /**
         * 执行任务
         *
         * @param idx
         */
        private void taskLoopRun(int idx) {
            Seat seat = seats[idx];
            Bucket next0 = seat.getHead();
            while (next0 != null) {
                if (next0.round <= 0) {
                    System.out.println("idx ==" + idx);
                    next0.task.run();
                    appends.incrementAndGet();
                    seat.remove(next0);
                } else {
                    next0.decrRound();
                }

                next0 = next0.next;
            }
        }

    }


}
